package S;

import U.Utils;
import rx.Observable;
import rx.Subscriber;
import rx.Subscription;
import rx.functions.Action1;
import rx.functions.Func2;

public class S {
    public static void main(String[] args) {
//        share();
//        scan();
        startWith();
        switchOnNext();
    }

    private static void switchOnNext() {

    }

    /**
     * startWith 实际是调用了concat 来保证各个Observable事件发射先后顺序
     */
    private static void startWith() {
        Observable starter = Observable.create(new Observable.OnSubscribe<String>() {

            public void call(Subscriber<? super String> subscriber) {
                subscriber.onNext("starter");
                subscriber.onCompleted();
            }
        });

        Observable creator = Observable.create(new Observable.OnSubscribe<String>() {

            public void call(Subscriber<? super String> subscriber) {
                    subscriber.onNext("Creator");
                    subscriber.onCompleted();
            }
        });
        Observable servant = Observable.create(new Observable.OnSubscribe<String>() {
            public void call(Subscriber<? super String> subscriber) {
                subscriber.onNext("servant");
                subscriber.onCompleted();
            }
        });
        servant.startWith(starter).startWith(creator).subscribe(new Action1() {
            public void call(Object o) {
                System.out.println(o);
            }
        });
    }

    /**
     * 把一个普通的observable转化成ConnectableObservable,将所有产生的事件会发射给所有的订阅者，
     * 并且自动取消订阅 ，每次产生的新数据都会发射。#{@link Observable#reduce(Func2)}操作符与之非常像，但是只会发射最后一个
     * 数据
     */
    public static void share() {
        Observable<Integer> share =
                Observable
                        .create(new Observable.OnSubscribe<Integer>() {

                            public void call(Subscriber<? super Integer> subscriber) {
                                for (int i = 0; i < 2; i++) {
                                    subscriber.onNext(i);
                                }
                                subscriber.onCompleted();
                            }
                        }).share();

        Subscription subscription = share.subscribe(new Action1<Integer>() {
            public void call(Integer integer) {
                System.out.println("s1: " + integer);
            }
        });


        Utils.sleep(1500);
        Subscription subscription1 = share.subscribe(new Action1<Integer>() {
            public void call(Integer integer) {
                System.out.println("s2: " + integer);
            }
        });

        System.out.println(subscription.isUnsubscribed());
        System.out.println(subscription1.isUnsubscribed());

    }

    /**
     * 把前一个产生的数据应用到后面的数据上产生一个新的数据，然后把这个新数据，又应用到后面一个数据上产生新数据，依次往复，
     * 直到应用到最后一个数据上
     */
    public static void scan() {
        System.out.println("==================scan===============");
        //计算阶乘
        Observable<Integer> ob = Observable.create(new Observable.OnSubscribe<Integer>() {
            public void call(Subscriber<? super Integer> subscriber) {
                for (int i = 1; i <= 5; i++) {
                    subscriber.onNext(i);
                }
                subscriber.onCompleted();
            }
        });
        ob.scan(new Func2<Integer, Integer, Integer>() {
            public Integer call(Integer integer, Integer integer2) {
                return integer * integer2;
            }
        }).subscribe(new Action1<Integer>() {
            public void call(Integer integer) {
                System.out.println(integer);
            }
        });

        //计算和
       /* Utils.sleep(120);
        ob.scan(new Func2<Integer, Integer, Integer>() {
            public Integer call(Integer integer, Integer integer2) {
                return integer + integer2;
            }
        }).subscribe(new Action1<Integer>() {
            public void call(Integer integer) {
                System.out.println(integer);
            }
        });*/
    }

    /**
     * 规范订阅者的行为，比如在多线程的情况下，订阅者的行为会发生混乱，调用了onComplete 但紧接着调用onNext，
     * 这显然不符合正确的订阅者该有的行为。
     */
    public static void serialize() {
        System.out.println("===============serialize==========");
        Observable.create(new Observable.OnSubscribe<Integer>() {
            public void call(Subscriber<? super Integer> subscriber) {
                for (int i = 0; i < 33; i++) {
                    subscriber.onNext(i);
                    if (i == 3)
                        subscriber.onCompleted();
                }
                subscriber.onCompleted();
            }
        }).serialize().subscribe(new Action1<Integer>() {
            public void call(Integer integer) {
                System.out.println(integer);
            }
        });//最终只会打印 0 1 2
    }
}
